package spark.example

import org.apache.spark.{SparkConf,SparkContext}

object MySort {
  implicit val playOrding = new Ordering[player1] {
    override def compare(x : player1, y: player1) : Int = {
      if(x.sortItema > y.sortItema) 1
      else if (x.sortItema > y.sortItema) {
        if(x.sortItemb > y.sortItemb) 1 else -1
      } else -1
    }
  }
}

object selfSortRule {
  def main(args: Array[String]) {
    val conf = new SparkConf().setAppName("selfSortRule example")
    val sc = new SparkContext(conf)
    val sortObject = sc.parallelize(Array(("Abc",60,110),("Bcd",70,50),("Def",60,90),("Cde",60,110),("Ghi",80,40),("Hij",90,30)))
    println("1==========")
    println("first self sort rule1,takeOrdered:")
    println("sort by 2nd element")
    println(sortObject.takeOrdered(sortObject.collect.size)(Ordering.by[(String,Int,Int),Int](_._2)).mkString(" "))
    println("sort by 2nd element,than 3nd")
    println(sortObject.takeOrdered(sortObject.collect.size)(Ordering.by[(String,Int,Int),(Int,Int)](x => (x._2,x._3)))mkString(" "))
    println()

    println("2==========")
    println("second self sort rule2 sortBy(Ordering) match case: ")
    println("descending order")
    println(sortObject.sortBy(x => player(x._2,x._3),false).collect.mkString(" "))
    
    println("ascending order")
    println(sortObject.sortBy(x => player(x._2,x._3)).collect.mkString(" "))
    println(sortObject.sortBy(x => player(x._2,x._3),true).collect.mkString(" "))
    println()

    println("3==========")
    import MySort._
    println("third self sort rule3,sortBy(Ordering) implicit transform: ")
    println("descending order")
    println(sortObject.sortBy(x => player1(x._2,x._3),false).collect.mkString(" "))
    println("ascending order")
    println(sortObject.sortBy(x => player1(x._2,x._3)).collect.mkString(" "))
    println(sortObject.sortBy(x => player1(x._2,x._3),true).collect.mkString(" "))
    sc.stop()

  }

  case class player(val sortItem1: Int,val sortItem2: Int) extends Ordered[player] with Serializable {
    override def compare(that: player) : Int = {
      if(this.sortItem1 == this.sortItem1) {
        this.sortItem2 - that.sortItem2
      } else {
        this.sortItem1 - that.sortItem1
      }
    }
  }

}

case class player1(sortItema : Int,sortItemb : Int) extends Serializable
